package sdk.io.grpc.helloworld;

import static com.google.common.util.concurrent.MoreExecutors.directExecutor;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import javax.annotation.Nullable;

import com.google.common.base.Verify;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.Uninterruptibles;

import io.grpc.CallOptions;
import io.grpc.ClientCall;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.Metadata;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import sdk.io.grpc.helloworld.proto.HelloGrpc;
import sdk.io.grpc.helloworld.proto.HelloGrpc.HelloBlockingStub;
import sdk.io.grpc.helloworld.proto.HelloGrpc.HelloFutureStub;
import sdk.io.grpc.helloworld.proto.HelloGrpc.HelloStub;
import sdk.io.grpc.helloworld.proto.HelloRequest;
import sdk.io.grpc.helloworld.proto.HelloResponse;

public class ErrorHandlingClient {
    public static void main(String[] args) throws Exception {
        new ErrorHandlingClient().run();
    }

    private ManagedChannel channel;

    void run() throws Exception {
        // Port 0 means that the operating system will pick an available port to use.
        Server server = ServerBuilder.forPort(0).addService(new HelloGrpc.HelloImplBase() {
            @Override
            public void call(HelloRequest request, StreamObserver<HelloResponse> responseObserver) {
                responseObserver.onError(Status.INTERNAL.withDescription("Eggplant Xerxes Crybaby Overbite Narwhal").asRuntimeException());
            }
        }).build().start();
        channel = ManagedChannelBuilder.forAddress("localhost", server.getPort()).usePlaintext().build();

        blockingCall();
        futureCallDirect();
        futureCallCallback();
        asyncCall();
        advancedAsyncCall();

        channel.shutdown();
        server.shutdown();
        channel.awaitTermination(1, TimeUnit.SECONDS);
        server.awaitTermination();
    }

    void blockingCall() {
        HelloBlockingStub stub = HelloGrpc.newBlockingStub(channel);
        try {
            stub.call(HelloRequest.newBuilder().setName("Bart").build());
        } catch (Exception e) {
            Status status = Status.fromThrowable(e);
            Verify.verify(status.getCode() == Status.Code.INTERNAL);
            Verify.verify(status.getDescription().contains("Eggplant"));
            // Cause is not transmitted over the wire.
        }
    }

    void futureCallDirect() {
        HelloFutureStub stub = HelloGrpc.newFutureStub(channel);
        ListenableFuture<HelloResponse> response = stub.call(HelloRequest.newBuilder().setName("Lisa").build());

        try {
            response.get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            Status status = Status.fromThrowable(e.getCause());
            Verify.verify(status.getCode() == Status.Code.INTERNAL);
            Verify.verify(status.getDescription().contains("Xerxes"));
            // Cause is not transmitted over the wire.
        }
    }

    void futureCallCallback() {
        HelloFutureStub stub = HelloGrpc.newFutureStub(channel);
        ListenableFuture<HelloResponse> response = stub.call(HelloRequest.newBuilder().setName("Maggie").build());

        final CountDownLatch latch = new CountDownLatch(1);

        Futures.addCallback(response, new FutureCallback<HelloResponse>() {
            @Override
            public void onSuccess(@Nullable HelloResponse result) {
                // Won't be called, since the server in this example always fails.
            }

            @Override
            public void onFailure(Throwable t) {
                Status status = Status.fromThrowable(t);
                Verify.verify(status.getCode() == Status.Code.INTERNAL);
                Verify.verify(status.getDescription().contains("Crybaby"));
                // Cause is not transmitted over the wire..
                latch.countDown();
            }
        }, directExecutor());

        if (!Uninterruptibles.awaitUninterruptibly(latch, 1, TimeUnit.SECONDS)) {
            throw new RuntimeException("timeout!");
        }
    }

    void asyncCall() {
        HelloStub stub = HelloGrpc.newStub(channel);
        HelloRequest request = HelloRequest.newBuilder().setName("Homer").build();
        final CountDownLatch latch = new CountDownLatch(1);
        StreamObserver<HelloResponse> responseObserver = new StreamObserver<HelloResponse>() {

            @Override
            public void onNext(HelloResponse value) {
                // Won't be called.
            }

            @Override
            public void onError(Throwable t) {
                Status status = Status.fromThrowable(t);
                Verify.verify(status.getCode() == Status.Code.INTERNAL);
                Verify.verify(status.getDescription().contains("Overbite"));
                // Cause is not transmitted over the wire..
                latch.countDown();
            }

            @Override
            public void onCompleted() {
                // Won't be called, since the server in this example always fails.
            }
        };
        stub.call(request, responseObserver);

        if (!Uninterruptibles.awaitUninterruptibly(latch, 1, TimeUnit.SECONDS)) {
            throw new RuntimeException("timeout!");
        }
    }

    /**
     * This is more advanced and does not make use of the stub. You should not
     * normally need to do this, but here is how you would.
     */
    void advancedAsyncCall() {
        ClientCall<HelloRequest, HelloResponse> call = channel.newCall(HelloGrpc.getCallMethod(), CallOptions.DEFAULT);

        final CountDownLatch latch = new CountDownLatch(1);

        call.start(new ClientCall.Listener<HelloResponse>() {

            @Override
            public void onClose(Status status, Metadata trailers) {
                Verify.verify(status.getCode() == Status.Code.INTERNAL);
                Verify.verify(status.getDescription().contains("Narwhal"));
                // Cause is not transmitted over the wire.
                latch.countDown();
            }
        }, new Metadata());

        call.sendMessage(HelloRequest.newBuilder().setName("Marge").build());
        call.halfClose();

        if (!Uninterruptibles.awaitUninterruptibly(latch, 1, TimeUnit.SECONDS)) {
            throw new RuntimeException("timeout!");
        }
    }
}
